package com.example.kafka.util;

import org.apache.kafka.common.errors.RetriableException;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

/**
 * Kafka异常与重试工具类
 */
public class KafkaRetryUtils {
    /**
     * 判断异常是否可重试
     */
    public static boolean isRetriable(Throwable e) {
        return e instanceof RetriableException;
    }

    /**
     * 带最大次数和退避的重试
     */
    public static <T> T retry(int maxRetries, long backoffMs, Supplier<T> action) throws Exception {
        int attempt = 0;
        while (true) {
            try {
                return action.get();
            } catch (Exception e) {
                if (++attempt > maxRetries || !isRetriable(e)) {
                    throw e;
                }
                TimeUnit.MILLISECONDS.sleep(backoffMs * attempt);
            }
        }
    }
}